package com.flute.haflute.agent.heartbeat;

import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;

import org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.transport.socket.nio.NioDatagramConnector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.agent.AgentContext;
import com.flute.haflute.agent.manage.ClusterResourceCenter;
import com.flute.haflute.agent.topo.ClusterTopoDBService;
import com.flute.haflute.agent.topo.Node;
import com.flute.haflute.agent.topo.RoleChangeListener;
import com.flute.haflute.common.ClusterConfLoader;
import com.flute.haflute.common.ClusterConstants.ROLE;
import com.flute.haflute.common.ClusterContext;
import com.flute.haflute.jobbox.base.RunProcessInterface;
import com.flute.haflute.net.HafluteKeepAliveFilterInMina;
import com.flute.haflute.net.messaging.SinkFilterInMina;

/**
 * 执行心跳巡检的后台线程; 非DM节点用于向DM定期发送心跳消息。
 */
public class HeartBeatGuard implements RoleChangeListener, RunProcessInterface {

	private static Logger logger = LoggerFactory
			.getLogger(HeartBeatGuard.class);

	/** sender 当前心跳发送处理器 */
	private HeartBeatSender sender = null;

	/** sendersMap 存储发送心跳消息的处理器的实例，用于角色变换时变换不同处理方式 */
	private Map<ROLE, HeartBeatSender> sendersMap;

	/** heartBeatConnector 发送心跳消息的udp连接 */
	private NioDatagramConnector heartBeatConnector;

	/** heartBeatSession 发送消息的session */
	private IoSession heartBeatSession;

	private boolean initialized = false;

	public HeartBeatGuard() {
		sendersMap = new HashMap<ROLE, HeartBeatSender>();
		sender = getActualSender(AgentContext.getNodeRole());
		logger.info("HeartBeat Guard started");
	}

	/**
	 * 建立到DM的连接, 使用MINA
	 */
	private void connectToDM() {
		heartBeatConnector = new NioDatagramConnector();
		heartBeatConnector.setConnectTimeoutMillis(60000L);
		heartBeatConnector.setConnectTimeoutCheckInterval(10000);
		heartBeatConnector.setHandler(sender);

		DefaultIoFilterChainBuilder chain = heartBeatConnector.getFilterChain();
		chain.addLast("keep-alive", new HafluteKeepAliveFilterInMina());
		chain.addLast("toMessageTyep", new SinkFilterInMina());
		if (logger.isDebugEnabled()) {
			chain.addLast("logger", new LoggingFilter());
		}

		ConnectFuture connFuture = heartBeatConnector
				.connect(new InetSocketAddress(ClusterContext.getDM(),
						ClusterContext.getHeartBeatPort()));
		connFuture.awaitUninterruptibly();
		heartBeatSession = connFuture.getSession();
	}

	/**
	 * 启动后台线程相关服务
	 */
	private void startup() {
		if (AgentContext.getNodeRole() != ROLE.DM) {
			connectToDM();
			new HeartBeatTask("HeartBeat Task").start();
		}
		initialized = true;
	}

	@Override
	public void shutdown() {
		if (heartBeatConnector != null && heartBeatConnector.isActive()) {
			heartBeatConnector.dispose();
		}
	}

	/*
	 * @see
	 * com.flute.haflute.agent.RoleChangeListener#roleChanged(com.flute.haflute
	 * .agent.ClusterRole)
	 */
	@Override
	public ROLE roleChanged(ROLE role) {
		sender = getActualSender(role);
		heartBeatConnector.setHandler(sender);
		return role;
	}

	private synchronized HeartBeatSender getActualSender(ROLE role) {
		HeartBeatSender sender = sendersMap.get(role);
		if (sender == null) {
			if (role == ROLE.DM) {
				sender = new DMHeartBeatSender();
			} else if (role == ROLE.BDM) {
				sender = new BDMHeartBeatSender();
			} else {
				sender = new SlaveHeartBeatSender();
			}
			sendersMap.put(role, sender);
		}
		return sender;
	}

	/**
	 * 心跳定期巡逻执行器，定期检查和发送心跳
	 * 
	 */
	class HeartBeatTask extends Thread {
		private long waitTime;
		private Random rand;

		public HeartBeatTask(String name) {
			super(name);
			waitTime = ClusterContext.getHeartBeatInterval();
			rand = new Random(waitTime);
		}

		@Override
		public void run() {
			while (true) {
				try {
					for (int i = 0; i < ClusterContext.getHeartBeatTries(); i++) {
						if (!heartBeatSession.isConnected()) {
							heartBeatConnector.dispose();
							connectToDM();
						}
						sender.sendHeartBeat(heartBeatSession);
						int nRound = 0;
						while (!sender.isSuccessful() && nRound < 50) {
							synchronized (sender) {
								sender.wait(200);
							}
							nRound++;
						}

						if (!sender.isSuccessful()) {
							ClusterResourceCenter.removeNode(new Node(
									ClusterContext.getDM()));
						} else {
							logger.debug("send one heartbeat to DM successfully");
							ClusterTopoDBService.getInstance().setDM(
									new Node(ClusterContext.getDM()));
							break;
						}
					}

					wait4NextRound();
				} catch (Exception e) {
					throw new RuntimeException(e);
				}
			}
		}

		private void wait4NextRound() {
			try {
				sleep(waitTime + rand.nextLong() % (waitTime / 10));
			} catch (InterruptedException e) {
			}
		}
	}

	public static void main(String[] args) {
		try {
			ClusterConfLoader.loadGlobalContext();
		} catch (Exception e) {
			e.printStackTrace();
		}
		if (args.length > 0) {
			String role = args[0];
			if (role.equalsIgnoreCase("bdm")) {
				AgentContext.setNodeRole(ROLE.BDM);
			}
		}
		new HeartBeatGuard().startup();
	}

	@Override
	public void callMain(String... params) {
		startup();
	}

	@Override
	public boolean isInitialized() {
		return initialized;
	}

	@Override
	public void restart() {

	}
}